package com.zhoulz.homework.chaptor02.zk;

import java.io.File;
import java.io.FileOutputStream;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.api.DeleteBuilder;
import org.apache.curator.framework.api.ExistsBuilder;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.api.SetDataBuilder;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.curator.utils.ZKPaths.PathAndNode;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;

import com.zhoulz.homework.chaptor02.config.Command;
import com.zhoulz.homework.chaptor02.config.Config;
import com.zhoulz.homework.chaptor02.node.Client;
import com.zhoulz.homework.chaptor02.node.CmdHelper;
import com.zhoulz.homework.chaptor02.node.Server;
import com.zhoulz.homework.chaptor02.utils.Util;

/**
 * zookeeper操作类
 * 
 * @author zhoulz
 *
 */
public class Zk{
	static Logger log = Logger.getLogger(Zk.class);
	// 应用目录
	public static final String APP = "/chapter02";
	// 节点配置目录
	public static final String APP_NODE_CONFIG = String.format("%s/%s/%s", APP, "config", "nodes");
	// 节点目录
	public static final String APP_NODE = "nodes";
	// 节点目录路径
	public static final String APP_NODE_PATH = String.format("%s/%s", APP, APP_NODE);
	// 命令目录
	public static final String APP_COMMANDS = "commands";
	// 命令目录路径
	public static final String APP_COMMANDS_PATH = String.format("%s/%s", APP, APP_COMMANDS);
	
	static PathChildrenCache nodeCache = null;
	static PathChildrenCache commandCache = null;
	
	public static SetDataBuilder setDataBuilder;
	public static GetDataBuilder getDateBuilder;
	public static GetChildrenBuilder getChildrenBuilder;
	public static ExistsBuilder existsBuilder;
	public static DeleteBuilder deleteBuilder;
	public static CreateBuilder createBuilder;
	
	/**
	 * 返回操作zk的curator客户端
	 * @param connectionString zk server连接串（host:port）
	 * @return
	 */
	public static CuratorFramework newClient(String connectionString) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);

        return CuratorFrameworkFactory.newClient(connectionString , 60 * 1000, 1 * 1000, retryPolicy);
	}
	
	/**
	 * 初始化。
	 * @param client
	 * @throws Exception
	 */
	public static void init(CuratorFramework client) throws Exception{
		setDataBuilder = client.setData();
		getDateBuilder = client.getData();
		getChildrenBuilder = client.getChildren();
		existsBuilder = client.checkExists();
		deleteBuilder = client.delete();
		createBuilder = client.create();
	}
	/**
	 * 单节点初始化。更新zk配置
	 * @param client
	 * @throws Exception
	 */
	public static void init(CuratorFramework client, String nodeName) throws Exception{
		init(client);
		
		String config = String.format("%s/%s", APP_NODE_CONFIG, nodeName);
		String localConfig = String.format("/%s.properties", Config.LOCAL_LEADER_CONFIG_NAME);
		
		// 使用zk上的配置
		byte[] b = getDateBuilder.forPath(config);
		log.info(String.format("使用zk配置(%s)，并更新本地配置(%s)", config, localConfig));
		FileOutputStream fo = new FileOutputStream(new File(Zk.class.getResource(localConfig).getFile()));
		fo.write(b, 0, b.length);
		fo.close();
		// 重新加载配置
		Config.load(Config.LOCAL_LEADER_CONFIG_NAME);
		
	}

	public static void initCache(CuratorFramework client, Server server) throws Exception {
		// 配置子节点目录监视器
		// 应用节点
		nodeCache = new PathChildrenCache(client, APP_NODE_PATH, true);
		nodeCache.start();
		addListener(nodeCache, server);
		
		// 命令节点
		commandCache = new PathChildrenCache(client, APP_COMMANDS_PATH, true);
		commandCache.start();
		addListener(commandCache, server);
	}
	/**
	 * 节点注册
	 * @param client
	 * @param nodeName 节点名
	 */
	public static void registerNode(Server server, String nodeName){
		try {
			// 注册为一个临时的节点,并将ip和端口写入数据
			String path = String.format("%s/%s", APP_NODE_PATH, nodeName);
			byte[] data = server.getHost().concat(":").concat(String.valueOf(server.getPort())).getBytes();
			// 如果存在，则删除
			if(existsBuilder.forPath(path) != null){
				// 保证删除
				deleteBuilder.guaranteed().forPath(path);
			}
			createBuilder
			.withMode(CreateMode.EPHEMERAL)
			.forPath(path, data);
			
		} catch (Exception e) {
			log.error("节点注册失败", e);
		}
	}
    private static void addListener(final PathChildrenCache cache,final Server server)
    {
        PathChildrenCacheListener listener = new PathChildrenCacheListener(){
			public void childEvent(CuratorFramework client,
					PathChildrenCacheEvent event) throws Exception {
				ChildData childData = event.getData();
				PathAndNode pathAndNode = ZKPaths.getPathAndNode(childData.getPath());
				String path = pathAndNode.getPath(),
						nodeName = pathAndNode.getNode();
                switch ( event.getType() )
                {
                    case CHILD_ADDED:
                    {
                    	if(path.endsWith(APP_NODE)){
                    		System.out.println("Node Added: " + ZKPaths.getNodeFromPath(childData.getPath()));
                    		// 应用目录
                    		if(server.getName().compareTo(nodeName) < 0){
                    			sendFile(server, nodeName, event.getData().getData());
                    		}
                    	}else if(path.endsWith(APP_COMMANDS)){
                    		doCommand(childData, nodeName);
                    	}
                        break;
                    }
                    case CHILD_UPDATED:
                    {
                    	if(path.endsWith(APP_COMMANDS)){
                    		doCommand(childData, nodeName);
                    	}
                        break;
                    }
                    case CHILD_REMOVED:
                    {
                        log.info("Node removed: " + nodeName);
                        break;
                    }
                }
			}

			private void doCommand(ChildData childData, String nodeName) throws Exception {
				// 命令目录
				byte[] commandData = childData.getData();
				String serverCommandNode = APP_COMMANDS.concat("-").concat(server.getName());
				// 匹配命令节点格式为命令前缀+节点名，如： commands-leader-1
				if(nodeName.startsWith(serverCommandNode)){
					String command = new String(commandData, Config.ENCODING);
					if(!Util.empty(command)){
						command = command.trim();
						if(Command.LSIT_FILE.equalsIgnoreCase(command)){
							// 查询接受到的文件信息，通过写入对应command节点，command窗口监视对应节点写回的数据
							// TODO 更好一点的方式，是通过tcp的方式发给command窗口
							setDataBuilder.forPath(childData.getPath(), server.getReceiveFiles().getBytes(Config.ENCODING));
						}else if(Command.SHUTDOWN.equalsIgnoreCase(command)){
							// 移除所有zk监听
							cache.getListenable().clear();
							nodeCache.close();
							commandCache.close();
							// 删除shutdown命令，防止第二次启不起来
//							setDataBuilder.forPath(childData.getPath(), "".getBytes());
							// 节点安全关闭
							CmdHelper.shutdown();
						}
					}
				}
			}
        };
        cache.getListenable().addListener(listener);
    }

	private static void sendFile(final Server server, String nodeName, byte[] nodeData) throws Exception {
		// Name小的连接大的，并发送随机文件
		log.info(server.getName().
				concat("发现节点").
				concat(nodeName).
				concat("已上线,启动TCP客户端发送数据文件！"));
		String[] serverCfgs = new String(nodeData).split(":");
		Client tcpClient = new Client(server.getName(), nodeName, serverCfgs[0], Integer.parseInt(serverCfgs[1]));
		tcpClient.start();
	}
}
